#include "service/DataCollector.h"
#include "device/DeviceManager.h"
#include "core/Logger.h"

#include <chrono>
#include <thread>
#include <cstdio>
#include <mutex>

namespace ems {
namespace service {

DataCollector::DataCollector()
    : running_(false), collectionInterval_(5), deviceManager_(nullptr),
      totalCollections_(0), successfulCollections_(0), failedCollections_(0),
      lastSuccessfulCollectionTs_(0), totalPointsCollected_(0) {
}

DataCollector::~DataCollector() {
    stop();
}

bool DataCollector::initialize(device::DeviceManager* deviceManager) {
    if (!deviceManager) {
        LOG_ERROR("DeviceManager cannot be null");
        return false;
    }

    deviceManager_ = deviceManager;
    LOG_INFO("DataCollector initialized with collection interval: " + std::to_string(collectionInterval_) + " seconds");
    return true;
}

void DataCollector::start() {
    if (running_) {
        LOG_WARN("DataCollector is already running");
        return;
    }

    if (!deviceManager_) {
        LOG_ERROR("DeviceManager not initialized, cannot start data collection");
        return;
    }

    LOG_INFO("Starting data collector with interval: " + std::to_string(collectionInterval_) + " seconds");
    running_ = true;
    collectorThread_ = std::thread(&DataCollector::collectionLoop, this);
}

void DataCollector::stop() {
    if (!running_) {
        return;
    }

    LOG_INFO("Stopping data collector");
    running_ = false;
    sleepCv_.notify_all();

    if (collectorThread_.joinable()) {
        collectorThread_.join();
    }

    LOG_INFO("Data collector stopped. Statistics - Total: " + std::to_string(totalCollections_) +
             ", Successful: " + std::to_string(successfulCollections_) +
             ", Failed: " + std::to_string(failedCollections_));
}

void DataCollector::collectionLoop() {
    LOG_INFO("Data collection loop started");
    try {
        while (running_) {
            uint64_t loopStartTime = getCurrentTimestamp();

        try {
            // Get all connected devices from DeviceManager
            auto connectedDevices = deviceManager_->getConnectedDevices();

            if (connectedDevices.empty()) {
                LOG_DEBUG("No connected devices found for data collection");
            } else {
                LOG_DEBUG("Collecting data from " + std::to_string(connectedDevices.size()) + " connected devices");

                // 按每设备采集间隔(poll_interval)调度：到期才触发采集
                uint64_t nowMs = getCurrentTimestamp();
                for (auto* device : connectedDevices) {
                    if (!running_) break; // Check if we should stop

                    if (device && device->isEnabled()) {
                        int intervalMs = device->getPollInterval();
                        if (intervalMs <= 0) intervalMs = 5000; // 默认5s
                        uint64_t lastMs = device->getLastCollectionTime();
                        bool due = (lastMs == 0) || (nowMs - lastMs >= static_cast<uint64_t>(intervalMs));
                        if (due) {
                            collectFromDevice(device);
                            totalCollections_++;
                        }
                    }
                }
            }
        } catch (const std::exception& e) {
            LOG_ERROR("Exception in data collection loop: " + std::string(e.what()));
        }

        // Calculate sleep time to maintain the desired collection interval
        uint64_t loopEndTime = getCurrentTimestamp();
        uint64_t loopDuration = loopEndTime - loopStartTime;
        uint64_t targetInterval = collectionInterval_ * 1000; // Convert to milliseconds

            if (loopDuration < targetInterval) {
                uint64_t sleepTime = targetInterval - loopDuration;
                std::unique_lock<std::mutex> lock(sleepMutex_);
                sleepCv_.wait_for(lock,
                                  std::chrono::milliseconds(sleepTime),
                                  [this]() { return !running_; });
            } else {
                LOG_WARN("Data collection loop took " + std::to_string(loopDuration) +
                         "ms, longer than target interval " + std::to_string(targetInterval) + "ms");
            }
        }

        LOG_INFO("Data collection loop stopped");
    } catch (const std::system_error& e) {
        std::fprintf(stderr, "[DataCollector::collectionLoop] system_error: %s\n", e.what());
        throw;
    }
}

bool DataCollector::collectFromDevice(device::DeviceBase* device) {
    if (!device) {
        LOG_ERROR("Device pointer is null");
        failedCollections_++;
        return false;
    }

    try {
        LOG_DEBUG("Collecting data from device: " + device->getName() + " (ID: " + device->getId() + ")");

        std::vector<device::DataValue> data;
        bool success = device->collectData(data);

        if (success) {
            successfulCollections_++;
            totalPointsCollected_.fetch_add(static_cast<uint64_t>(data.size()), std::memory_order_relaxed);
            lastSuccessfulCollectionTs_.store(getCurrentTimestamp(), std::memory_order_relaxed);
            LOG_DEBUG("Successfully collected " + std::to_string(data.size()) +
                     " data points from device: " + device->getName());

            // Call the data callback if registered
            if (dataCallback_) {
                try {
                    dataCallback_(device->getId(), data);
                } catch (const std::exception& e) {
                    LOG_ERROR("Exception in data callback for device " + device->getId() + ": " + e.what());
                }
            }

            // Log some sample data for debugging (first few data points)
            for (size_t i = 0; i < std::min(data.size(), size_t(3)); ++i) {
                const auto& dataValue = data[i];
                LOG_DEBUG("  " + dataValue.key + ": " + std::to_string(dataValue.value) +
                         (dataValue.unit.empty() ? "" : " " + dataValue.unit));
            }

            if (data.size() > 3) {
                LOG_DEBUG("  ... and " + std::to_string(data.size() - 3) + " more data points");
            }

        } else {
            failedCollections_++;
            std::string err = device->getLastError();
            LOG_WARN("Failed to collect data from device: " + device->getName() +
                    " - " + err);

            if (errorCallback_) {
                try {
                    errorCallback_(device->getId(), err);
                } catch (const std::exception& e) {
                    LOG_ERROR("Exception in data error callback for device " + device->getId() + ": " + e.what());
                }
            }
        }

        return success;

    } catch (const std::exception& e) {
        failedCollections_++;
        LOG_ERROR("Exception collecting data from device " + device->getName() + ": " + e.what());
        return false;
    }
}

uint64_t DataCollector::getCurrentTimestamp() const {
    auto now = std::chrono::system_clock::now();
    auto duration = now.time_since_epoch();
    return std::chrono::duration_cast<std::chrono::milliseconds>(duration).count();
}

}  // namespace service
}  // namespace ems
